package com.xzx.flink.streamapi.state;

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.StringJoiner;

/**
 * 广播状态
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/16 11:56
 */
public class BroadcastState_01_Demo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 读取用户行为事件流
        DataStreamSource<Action> actionStream = env.fromElements(new Action("Alice", "login"), new Action("Alice", "pay"), new Action("Bob", "login"), new Action("Bob", "buy"));

        // 定义行为模式流，代表了要检测的标准
        DataStreamSource<Pattern> patternStream = env.fromElements(new Pattern("login", "pay"), new Pattern("login", "buy"));
        MapStateDescriptor<Void, Pattern> patternMapStateDescriptor = new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(Pattern.class));
        BroadcastStream<Pattern> ruleStream = patternStream.broadcast(patternMapStateDescriptor);

        actionStream.keyBy(Action::getUserId).connect(ruleStream).process(new KeyedBroadcastProcessFunction<String, Action, Pattern, Tuple2<String, Pattern>>() {

            // 记录一下上一次的行为
            private ValueState<String> previousActionState;

            @Override
            public void open(Configuration parameters) throws Exception {
                previousActionState = getRuntimeContext().getState(new ValueStateDescriptor<>("previousAction", String.class));
            }

            @Override
            public void processElement(Action value, KeyedBroadcastProcessFunction<String, Action, Pattern, Tuple2<String, Pattern>>.ReadOnlyContext ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
                String previousAction = this.previousActionState.value();
                MapStateDescriptor<Void, Pattern> patternMapStateDescriptor = new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(Pattern.class));
                Pattern pattern = ctx.getBroadcastState(patternMapStateDescriptor).get(null);
                if (previousAction != null && pattern != null) {
                    if (pattern.action1.equals(previousAction) && pattern.action2.equals(value.getAction())) {
                        out.collect(Tuple2.of(ctx.getCurrentKey(), pattern));
                    }
                }
                previousActionState.update(value.getAction());
            }

            @Override
            public void processBroadcastElement(Pattern value, KeyedBroadcastProcessFunction<String, Action, Pattern, Tuple2<String, Pattern>>.Context ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
                //广播规则
                MapStateDescriptor<Void, Pattern> patternMapStateDescriptor = new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(Pattern.class));
                ctx.getBroadcastState(patternMapStateDescriptor).put(null, value);
            }

        }).print();

        env.execute(BroadcastState_01_Demo.class.getSimpleName());
    }

    // 定义行为模式 POJO 类，包含先后发生的两个行为
    public static class Pattern {
        private String action1;
        private String action2;

        public Pattern() {
        }

        public Pattern(String action1, String action2) {
            this.action1 = action1;
            this.action2 = action2;
        }

        public String getAction1() {
            return action1;
        }

        public void setAction1(String action1) {
            this.action1 = action1;
        }

        public String getAction2() {
            return action2;
        }

        public void setAction2(String action2) {
            this.action2 = action2;
        }

        @Override
        public String toString() {
            return new StringJoiner(", ", Pattern.class.getSimpleName() + "[", "]").add("action1='" + action1 + "'").add("action2='" + action2 + "'").toString();
        }
    }

    public static class Action {
        private String userId;
        private String action;

        public Action() {
        }

        public Action(String userId, String action) {
            this.userId = userId;
            this.action = action;
        }

        public String getUserId() {
            return userId;
        }

        public void setUserId(String userId) {
            this.userId = userId;
        }

        public String getAction() {
            return action;
        }

        public void setAction(String action) {
            this.action = action;
        }

        @Override
        public String toString() {
            return new StringJoiner(", ", Action.class.getSimpleName() + "[", "]").add("userId='" + userId + "'").add("action='" + action + "'").toString();
        }
    }
}
